package com.gitee.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/*
  非并行数据源:
    整个数据只有一个n+1
  前提: 不是一个并行source
  1. 实现SourceFunction接口, cancel()方法在flink任务呗取消时执行, run()方法是用来生成数据的
  2. 将实现SourceFunction的对象传递给senv
 */
object CustomerSourceWithNoParallel {
  def main(args: Array[String]): Unit = {
    //1.创建senv
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    senv.setParallelism(3)
    //2.从自定义的数据源读取数据
    import org.apache.flink.api.scala._
    val data: DataStream[Long] = senv.addSource(new MyNoParallel())
    data.print()
    senv.execute()
  }

}

class MyNoParallel extends SourceFunction[Long]() {
  var currentNum: Long = 1
  var isRunning: Boolean = true

  //表示是否正在执行
  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
    while (isRunning) {
      //使用ctx将数据生成出去
      ctx.collect(currentNum)
      currentNum += 1
      Thread.sleep(1000)
    }
  }
}
